package org.xiangxji.util.framework.threadpool;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.xiangxji.util.framework.util.NamedThreadFactory;


/**
 * 简单的任务分发类，支持根据线程权重模型来分配资源，当前支持两类
 *  1. 根据某一类key可以预留多少资源独享。
 *  2. 根据某一类key可以限制最多使用多少资源。
 * 
 * @author xiangxji
 */
public class JobDispatcher extends Thread {

	private static final Log log = LogFactory.getLog(JobDispatcher.class);

	public static final String DEFAULT_COUNTER = "defaultCounter";
	public static final String TOTAL_COUNTER = "totalCounter";
	public static final String QUEUE_JOB_LENGTH = "queue_job_length";

	private BlockingQueue<Job> jobQueue;
	private JobThreadPoolExecutor threadPool;
	private JobThreadWeightModel[] jobThreadWeightModel;
	private Map<String, AtomicInteger> counterPool;
	private Map<String, Integer> thresholdPool;
	private int defaultThreshold;
	private AtomicInteger defaultCounter;
	private AtomicInteger totalCounter;
	private int maximumPoolSize = 200;
	private int maximumQueueSize = 1000;
	private boolean isRunning = true;

	public Map<String, Integer> getCurrentThreadStatus() {
		Map<String, Integer> status = new HashMap<String, Integer>();
		Iterator<Entry<String, AtomicInteger>> entrys = counterPool.entrySet().iterator();

		while (entrys.hasNext()) {
			Entry<String, AtomicInteger> e = entrys.next();
			status.put(e.getKey(), e.getValue().get());
		}

		status.put(DEFAULT_COUNTER, defaultCounter.get());
		status.put(QUEUE_JOB_LENGTH, jobQueue.size());
		status.put(TOTAL_COUNTER, totalCounter.get());
		return status;
	}
	
	

	public JobThreadPoolExecutor getThreadPool() {
		return threadPool;
	}

	public void setThreadPool(JobThreadPoolExecutor threadPool) {
		this.threadPool = threadPool;
	}


	public void init() {
		if (threadPool == null)
			threadPool = new JobThreadPoolExecutor(maximumPoolSize, 0L, TimeUnit.SECONDS,
				new LinkedBlockingQueue<Runnable>(maximumQueueSize), new NamedThreadFactory("jobDispatcher_worker"), this);

		defaultThreshold = maximumPoolSize;
		defaultCounter = new AtomicInteger(0);
		totalCounter = new AtomicInteger(0);
		jobQueue = new LinkedBlockingQueue<Job>(maximumQueueSize);
		counterPool = new ConcurrentHashMap<String, AtomicInteger>();
		thresholdPool = new HashMap<String, Integer>();

		buildWeightModel();
	}

	public void stopDispatcher() {
		if (threadPool != null)
			threadPool.shutdownNow();

		if (jobQueue != null)
			jobQueue.clear();

		if (counterPool != null)
			counterPool.clear();

		if (thresholdPool != null)
			thresholdPool.clear();

		isRunning = false;
		this.interrupt();
	}

	private void buildWeightModel() {
		if (jobThreadWeightModel != null && jobThreadWeightModel.length > 0) {
			for (JobThreadWeightModel j : jobThreadWeightModel) {
				try {
					counterPool.put(j.getKey(), new AtomicInteger(0));

					if (j.getValue() == 0) continue;

					if (j.getType().equals(JobThreadWeightModel.WEIGHT_MODEL_LIMIT)) {
						thresholdPool.put(j.getKey(), j.getValue());
					} else if (j.getType().equals(JobThreadWeightModel.WEIGHT_MODEL_LEAVE)) {
						thresholdPool.put(j.getKey(), j.getValue());
						defaultThreshold -= j.getValue();
					} else
						throw new RuntimeException("thread weight config not support!");
				} catch (Exception ex) {
					log.error("create jobWeightModels: " + j.getKey() + " error!", ex);
				}
			}

			if (defaultThreshold <= 0)
				throw new RuntimeException("total leave resource > total resource...");

			for (JobThreadWeightModel j : jobThreadWeightModel) {
				try {
					if (j.getValue() == 0)
						continue;

					if (j.getType().equals(JobThreadWeightModel.WEIGHT_MODEL_LIMIT)) {
						if (thresholdPool.get(j.getKey()) > defaultThreshold)
							thresholdPool.put(j.getKey(), -defaultThreshold);
						else
							thresholdPool.put(j.getKey(), -thresholdPool.get(j.getKey()));
					}
				} catch (Exception ex) {
					log.error("create jobWeightModels: " + j.getKey() + " error!", ex);
				}
			}
		}
	}

	/**
	 * 提交任务
	 */
	public void submitJob(Job job) {
		// 第一层做总量判断，同时锁定总资源
		if (totalCounter.incrementAndGet() > this.maximumPoolSize) {
			totalCounter.decrementAndGet();

			if (!jobQueue.offer(job)) {// 补偿job
				throw new RuntimeException("can't submit job, queue insert error.");
			}

			return;
		}

		String key = job.getKey();
		Integer threshold = thresholdPool.get(key);
		boolean hasResource = false;

		if (threshold == null) {
			if (defaultCounter.incrementAndGet() > defaultThreshold) {
				defaultCounter.decrementAndGet();
			} else {
				hasResource = true;
			}
		} else {
			AtomicInteger counter = counterPool.get(key);
			if (threshold > 0) {// leave mode
				if (counter.incrementAndGet() > threshold) {
					counter.decrementAndGet();
					if (defaultCounter.incrementAndGet() > defaultThreshold) {
						defaultCounter.decrementAndGet();
					} else {
						hasResource = true;
					}
				} else {
					hasResource = true;
				}
			} else {// limit mode
				if (counter.incrementAndGet() > -threshold) {
					counter.decrementAndGet();
				} else {
					if (defaultCounter.incrementAndGet() > defaultThreshold) {
						defaultCounter.decrementAndGet();
						counter.decrementAndGet();
					} else {
						hasResource = true;
					}
				}
			}
		}

		if (hasResource) {
			threadPool.execute(job);
		} else {
			totalCounter.decrementAndGet();
			if (!jobQueue.offer(job)) {// 补偿job
				throw new RuntimeException("can't submit job, queue full...");
			}
		}
	}

	public JobThreadWeightModel[] getJobThreadWeightModel() {
		return jobThreadWeightModel;
	}

	public void setJobThreadWeightModel(JobThreadWeightModel[] jobThreadWeightModel) {
		this.jobThreadWeightModel = jobThreadWeightModel;
	}

	public void setJobThreadWeightModel(List<JobThreadWeightModel> jobThreadWeightModel) {
		this.jobThreadWeightModel = jobThreadWeightModel.toArray(new JobThreadWeightModel[0]);
	}

	public BlockingQueue<Job> getJobQueue() {
		return jobQueue;
	}

	public void setJobQueue(BlockingQueue<Job> jobQueue) {
		this.jobQueue = jobQueue;
	}

	public int getMaximumPoolSize() {
		return maximumPoolSize;
	}

	public void setMaximumPoolSize(int maximumPoolSize) {
		this.maximumPoolSize = maximumPoolSize;
	}

	public int getMaximumQueueSize() {
		return this.maximumQueueSize;
	}

	public void setMaximumQueueSize(int maximumQueueSize) {
		this.maximumQueueSize = maximumQueueSize;
	}

	public Map<String, AtomicInteger> getCounterPool() {
		return counterPool;
	}

	public Map<String, Integer> getThresholdPool() {
		return this.thresholdPool;
	}

	public int getDefaultThreshold() {
		return defaultThreshold;
	}

	public AtomicInteger getDefaultCounter() {
		return defaultCounter;
	}

	public AtomicInteger getTotalCounter() {
		return totalCounter;
	}

	@Override
	public void run() {
		try {
			while (isRunning) {
				Job job = jobQueue.poll(1, TimeUnit.SECONDS);

				if (job != null) {
					submitJob(job);
				} else {
					Thread.sleep(100);
				}
			}
		} catch (InterruptedException ex) {
			// do nothing
		}
	}

}
